package org.opens.javaskill.thread.threadPool;

import java.util.ArrayList;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;

/**
 * @Date 2020/8/19 21:16
 * @Created by 99126
 */
public class SimpleThreadPool extends Thread {

    /**
     * 线程池大小
     */
    private int size;

    private final int taskQueueSize;

    private final static int DEFAULT_TASK_QUEUE_SIZE = 2000;

    /**
     * 任务队列(TaskQueue)
     */
    private final static LinkedList<Runnable> TASK_QUEUE = new LinkedList<>();

    /**
     * 所创建线程name的前缀
     */
    private final static String THREAD_PREFIX = "simple-thread-pool-";

    /**
     * 所创建的线程都加入一个ThreadGroup中, 方便管理
     */
    private final static ThreadGroup GROUP = new ThreadGroup("pool-group");

    /**
     * 线程命名自增
     */
    private static volatile int seq = 0;

    /**
     * 用来容纳当前活动线程的List
     */
    private final static List<WorkerTask> THREAD_QUEUE = new ArrayList<>();

    private final DiscardPolicy discardPolicy;

    private volatile boolean destroy = false;

    private int min;

    private int max;

    private int active;

    private final static int DEFAULT_MIN = 4;
    private final static int DEFAULT_ACTIVE = 8;
    private final static int DEFAULT_MAX = 12;


    public final static DiscardPolicy DEFAULT_DISCARD_POLICY = () -> {
        throw new DiscardException("默认决绝策略触发");
    };

    public SimpleThreadPool() {
        this(DEFAULT_MIN, DEFAULT_ACTIVE, DEFAULT_MAX, DEFAULT_TASK_QUEUE_SIZE, DEFAULT_DISCARD_POLICY);
    }

    public SimpleThreadPool(int min, int active, int max, int taskQueueSize, DiscardPolicy discardPolicy) {
        this.min = min;
        this.active = active;
        this.max = max;
        this.taskQueueSize = taskQueueSize;
        this.discardPolicy = discardPolicy;

        init();
    }

    /**
     * 执行初始化, 创建出这个线程池中的所有线程并启动.
     */
    private void init() {
        for (int i = 0; i < this.min; i++) {
            createWorkerTask();
        }
        this.size = min;
        this.start();
    }

    private void createWorkerTask() {
        WorkerTask task = new WorkerTask(GROUP, THREAD_PREFIX + (seq++));
        task.start();
        THREAD_QUEUE.add(task);
    }

    public void submit(Runnable runnable) {
        if ( destroy ) {
            throw new IllegalStateException("当前线程池已经销毁");
        }
        synchronized (TASK_QUEUE) {
            if ( TASK_QUEUE.size() > taskQueueSize ) {
                discardPolicy.discard();
            } else {
                TASK_QUEUE.addLast(runnable);
                TASK_QUEUE.notifyAll();
            }
        }
    }

    public void shutdown() throws InterruptedException {
        // 如果任务没有处理完, 就一直等待处理完
        while (!TASK_QUEUE.isEmpty()) {
            Thread.sleep(50);
        }
        synchronized (THREAD_QUEUE) {
            int workerThreadSize = THREAD_QUEUE.size();
            // 如果正在工作的线程集合不为空, 则开始打断那些处于BLOCKED状态的线程, 并等待
            while (workerThreadSize > 0) {
                for(WorkerTask workerTask : THREAD_QUEUE) {
                    if ( workerTask.getTaskState() ==  TaskState.BLOCKED ) {
                        // 打断会打断处于BLOCKED的线程, 如果
                        workerTask.interrupt();
                        // 如果打断时线程的代码不再wait所在的哪个代码区, 就需要改变taskState的状态来停止线程, 因为线程的最外层条件使taskState != TaskState.DEAD
                        workerTask.close();
                        // 每打断一个, 就使正在活动线程数量-1
                        workerThreadSize--;
                    } else {
                        Thread.sleep(50);
                    }
                }
            }
            this.destroy = true;
        }
        System.out.println("线程池关闭了");
    }

    @Override
    public void run() {
        while (!destroy) {
            System.out.printf("ThreadPool #min: %d, active: %d, max:%d, current: %d, queueSize: %d\n",
                    this.min, this.active, this.max, this.size, TASK_QUEUE.size());
            try {
                Thread.sleep(3000);
                // 线程扩容算法
                if ( TASK_QUEUE.size() > active && size < active ) {
                    for (int i = size; i < active; i++) {
                        createWorkerTask();
                    }
                    size = active;
                    System.out.println("线程扩容至active");
                } else if ( TASK_QUEUE.size() > max && size < max ) {
                    for (int i = size; i < max; i++) {
                        createWorkerTask();
                    }
                    size = max;
                    System.out.println("线程扩容至max");
                }

                // 线程缩容算法
                if ( TASK_QUEUE.isEmpty() && size > active ) {
                    System.out.println("任务队列为空, 开始缩容到active");
                    synchronized (THREAD_QUEUE) {
                        int releaseSize = size - active;
                        Iterator<WorkerTask> workerTaskIterator = THREAD_QUEUE.iterator();
                        while (workerTaskIterator.hasNext()) {
                            if ( releaseSize <= 0 ) {
                                break;
                            }
                            WorkerTask workerTask = workerTaskIterator.next();
                            workerTask.close();
                            workerTask.interrupt();
                            workerTaskIterator.remove();
                            releaseSize--;
                        }
                        size = active;
                    }
                } else if ( TASK_QUEUE.isEmpty() && size > min ) {
                    System.out.println("任务队列为空, 进一步缩容到min");
                    synchronized (THREAD_QUEUE) {
                        int releaseSize = size - min;
                        Iterator<WorkerTask> workerTaskIterator = THREAD_QUEUE.iterator();
                        while (workerTaskIterator.hasNext()) {
                            if ( releaseSize <= 0 ) {
                                break;
                            }
                            WorkerTask workerTask = workerTaskIterator.next();
                            workerTask.close();
                            workerTask.interrupt();
                            workerTaskIterator.remove();
                            releaseSize--;
                        }
                        size = min;
                    }
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

    public boolean isDestroy() {
        return this.destroy;
    }

    public int getSize() {
        return size;
    }

    public int getTaskQueueSize() {
        return taskQueueSize;
    }

    public int getMin() {
        return min;
    }

    public int getMax() {
        return max;
    }

    public int getActive() {
        return active;
    }

    public static class DiscardException extends RuntimeException {

        public DiscardException(String message) {
            super(message);
        }

    }

    public static interface DiscardPolicy {

        public void discard() throws DiscardException;

    }

    private static class WorkerTask extends Thread {

        /**
         * 保存线程状态
         */
        private volatile TaskState taskState = TaskState.FREE;

        /**
         * 创建时指定线程组和线程名称
         *
         * @param group 线程组
         * @param name  线程名称
         */
        public WorkerTask(ThreadGroup group, String name) {
            super(group, name);
        }

        /**
         * 获取线程状态
         */
        public TaskState getTaskState() {
            return this.taskState;
        }

        /**
         * 调整当前线程的状态
         */
        public void close() {
            this.taskState = TaskState.DEAD;
        }

        @Override
        public void run() {
            OUTER:
            while (this.taskState != TaskState.DEAD) {
                Runnable runnable = null;
                // 这里需要读取TaskQueue, 所以需要加锁
                synchronized (SimpleThreadPool.TASK_QUEUE) {
                    // 如果TaskQueue为空, 就使当前线程wait
                    while (SimpleThreadPool.TASK_QUEUE.isEmpty()) {
                        try {
                            taskState = TaskState.BLOCKED;
                            SimpleThreadPool.TASK_QUEUE.wait();
                        } catch (InterruptedException e) {
                            System.out.println(Thread.currentThread().getName() + " 被释放");
                            // 相当于中止了线程, 打断时直接终止这个线程
                            break OUTER;
                        }
                    }
                    runnable = SimpleThreadPool.TASK_QUEUE.removeFirst();
                }
                // 执行获取到的任务并执行
                if (runnable != null) {
                    // 在执行前调整线程的状态为RUNNING
                    taskState = TaskState.RUNNING;
                    runnable.run();
                    // 任务执行完毕之后调整为FREE
                    taskState = TaskState.FREE;
                }
            }
        }
    }

    /**
     * 线程状态枚举
     */
    private enum TaskState {

        FREE,       // 空闲
        RUNNING,    // 正在执行任务
        BLOCKED,    // 睡眠中, 执行了wait方法
        DEAD;       // 死亡了.

    }


}


class SimpleThreadPoolTest {

    public static void main(String[] args) throws InterruptedException {
        SimpleThreadPool simpleThreadPool = new SimpleThreadPool();
        for (int i = 0; i < 280; i++) {
            final int j = i;
            simpleThreadPool.submit(() -> {
                System.out.println(Thread.currentThread().getName() + " service for " + j + " start");
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println(Thread.currentThread().getName() + " service for " + j + " end");
            });
        }

        Thread.sleep(20000);
        simpleThreadPool.shutdown();
    }

}

